package main

import (
	"bytes"
	"encoding/json"
	"flag"
	"fmt"
	"io/ioutil"
	"log"
	"net/http"
	"os"
	"time"

	"github.com/olivere/elastic/v7"
)

/**
* Author: dinglin
* Date: 2021/8/31 10:32 上午
 */

func doReindex() {
	if config.fromHost == "" {
		log.Printf("from_host miss")
		flag.PrintDefaults()
		os.Exit(-1)
	}
	if config.fromIndex == "" || config.toIndex == "" {
		log.Printf("from_index or to_index or snapshot_name miss")
		flag.PrintDefaults()
		os.Exit(-1)
	}
	from, err := NewEsClient(config.fromHost)
	if err != nil {
		os.Exit(-1)
	}
	createIndex(from, config.toIndex)
	setIndexNumOfReplicas(from, config.toIndex, 0)
	err = reindex(from, config.fromHost, config.fromIndex, config.toIndex)
	if err != nil {
		os.Exit(-1)
	}
}

func reindex(client *elastic.Client, esHost, src, dest string) error {
	log.Printf("start reindex %s --> %s", src, dest)
	url := fmt.Sprintf("%s/_reindex?slices=9&refresh&wait_for_completion=false", esHost)
	body := map[string]interface{}{
		"source": map[string]interface{}{
			"index": src,
		},
		"dest": map[string]interface{}{
			"index":   dest,
			"op_type": "create",
		},
		"conflicts": "proceed",
	}

	data, _ := json.Marshal(body)
	resp, err := http.DefaultClient.Post(url, "application/json", bytes.NewReader(data))
	if err != nil {
		log.Printf("do req failed:%v", err)
		return err
	}
	defer resp.Body.Close()

	data, err = ioutil.ReadAll(resp.Body)
	if err != nil {
		log.Printf("end reindex %s --> %s: %v", src, dest, err)
		return err
	}
	log.Printf("reindex %s --> %s: %v", src, dest, string(data))

	log.Printf("wait reindex complete")
	counter := 0
	for range time.Tick(time.Minute) {
		success, err := getTask(client, "*reindex")
		if err != nil || !success {
			continue
		}
		if counter >= 4 {
			break
		}
		srcCnt := indexCount(client, src)
		destCnt := indexCount(client, dest)
		log.Printf("reindex %s[%d] --> %s[%d]", src, srcCnt, dest, destCnt)
		counter++
	}
	log.Printf("complete reindex %s --> %s: %v", src, dest, print(resp))
	return nil
}
